/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.storage.mem;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Vector;

import cgl.narada.event.TemplateInfo;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.reliable.CompanionDbInfo;
import cgl.narada.service.storage.InventoryEvent;
import cgl.narada.service.storage.SequenceDestinations;
import cgl.narada.service.storage.StorageService;

/** Provides a host of storage related operations that are important for
 ensuring reliable delivery

 @author Shrideep Pallickara
 $Date$
 $Revision$
 */

public class StorageServiceMemImpl implements StorageService {
    /** Maintains a list of all the registered entities */
    private Vector registeredEntities;

    /** Maintains a list of templates and the list of entities registered to this
     template. template-id's to a Hashtable of registered entities */
    private Hashtable templatesAndEntities;

    /** template-id's to templateInfo */
    private Hashtable templates;

    private String moduleName = "StorageServiceMemImpl: ";

    private Hashtable profiles;

    private Hashtable templatesAndProfiles;

    private Hashtable templatesAndSyncpoints;

    private Hashtable storedEvents;

    private Hashtable destinationTable;

    private Hashtable catenations;

    private Hashtable sequencesLastAssigned;

    private long lastSequenceNumber;

    public StorageServiceMemImpl() {
        registeredEntities = new Vector();
        templates = new Hashtable();
        templatesAndEntities = new Hashtable();
        profiles = new Hashtable();

        templatesAndProfiles = new Hashtable();
        templatesAndSyncpoints = new Hashtable();

        storedEvents = new Hashtable();
        destinationTable = new Hashtable();

        catenations = new Hashtable();

        sequencesLastAssigned = new Hashtable();
    }

    public void storeProfile(int templateId, Profile profile) throws
            ServiceException {
        Object templateKey = new Integer(templateId);

        profiles.put(profile.getProfileId(), profile);

        Hashtable profileList = (Hashtable) templatesAndProfiles.get(
                templateKey);
        profileList.put(profile.getProfileId(), profile);

    }

    public void removeProfile(int templateId, Profile profile) throws
            ServiceException {
        Object templateKey = new Integer(templateId);

        profiles.remove(profile.getProfileId());

        Hashtable profileList = (Hashtable) templatesAndProfiles.get(
                templateKey);
        profileList.remove(profile.getProfileId());

    }

    /** Stores a registered entity */
    public void storeRegisteredEntity(int entityId) throws ServiceException {
        Object entityKey = new Integer(entityId);
        registeredEntities.addElement(entityKey);
    }

    /** Removes a registered entity. This method should also remove the entity
     from any profiles that it was subscribed to and and templateId to which
     it was registered */
    public void removeRegisteredEntity(int entityId) throws ServiceException {
        Object entityKey = new Integer(entityId);
        registeredEntities.removeElement(entityKey);
    }

    public void storeTemplate(TemplateInfo templateInfo) throws
            ServiceException {
        Object templateKey = new Integer(templateInfo.getTemplateId());

        templates.put(templateKey, templateInfo);

        /** Initialize Hashtable for the management of registered entities */
        templatesAndEntities.put(templateKey, new Hashtable());

        templatesAndProfiles.put(templateKey, new Hashtable());
        templatesAndSyncpoints.put(templateKey, new Hashtable());
    }

    public void removeTemplateManagement(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);

        templates.remove(templateKey);
        /** Remove entries pertaining to the management of registered entities */
        templatesAndEntities.remove(templateKey);

        templatesAndSyncpoints.remove(templateKey);
    }

    public void storeEntityForTemplate(int entityId, int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        Object entityKey = new Integer(entityId);

        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);
        entityList.put(entityKey, entityKey);
    }

    /** Deregister's an entity from a managed template */
    public void removeEntityFromTemplate(int entityId, int templateId) throws
            ServiceException {
        Object entityKey = new Integer(entityId);
        Object templateKey = new Integer(templateId);

        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);
        entityList.remove(entityKey);
    }

    /** Returns a list of registered entities */
    public int[] getListOfRegisteredEntities() throws ServiceException {

        if (registeredEntities.size() == 0) {
            return null;
        }

        int[] list = new int[registeredEntities.size()];
        Enumeration e = registeredEntities.elements();

        int tracker = 0;
        while (e.hasMoreElements()) {
            list[tracker++] = ((Integer) e.nextElement()).intValue();
        }
        return list;
    }

    /** Returns a list of registered entities to a given template */
    public int[] getListOfRegisteredEntities(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);

        if (entityList.size() == 0) {
            return null;
        }

        int[] list = new int[entityList.size()];
        Enumeration e = entityList.elements();

        int tracker = 0;
        while (e.hasMoreElements()) {
            list[tracker++] = ((Integer) e.nextElement()).intValue();
        }
        return list;
    }

    /** Returns a list of profiles registered to a given template */
    public Profile[] getListOfRegisteredProfiles(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        Hashtable profileList = (Hashtable) templatesAndProfiles.get(
                templateKey);

        if (profileList.size() == 0) {
            return null;
        }

        return (Profile[]) profileList.values().toArray(new Profile[0]);

    }

    /**  Returns a array containing the list of managed templates */
    public TemplateInfo[] getListOfManagedTemplates() throws ServiceException {

        if (templates.size() == 0) {
            return null;
        }

        return (TemplateInfo[]) templates.values().toArray(new TemplateInfo[0]);
    }

    /** This method should not return unless  the event has been written to
     stable storage */
    public void store(CompanionDbInfo companionDbInfo,
                      InventoryEvent inventoryEvent,
                      SequenceDestinations sequenceDestinations) throws
            ServiceException {

        Object key = new Long(inventoryEvent.getSequenceNumber());

        Object templateKey = new Long(inventoryEvent.getTemplateId());
        long inventSeq = inventoryEvent.getSequenceNumber();

        if (lastSequenceNumber > inventSeq) {
            System.out.println(moduleName +
                               "The sequence number assigned is less"
                               + "than the last number previously assigned!");
            return;
        }

        long lastSequenceNumber = inventSeq;

        if (sequencesLastAssigned.containsKey(templateKey)) {
            sequencesLastAssigned.remove(templateKey);
        }

        sequencesLastAssigned.put(templateKey, new Long(lastSequenceNumber));

        storedEvents.put(key, inventoryEvent);
        destinationTable.put(key, sequenceDestinations);

    }

    /** Upon receipt of a NAK request we need to retrieve the inventory event
     that corresponds to  */
    public InventoryEvent getStoredEvent(long sequenceNumber) throws
            ServiceException {
        Object key = new Long(sequenceNumber);
        return (InventoryEvent) storedEvents.get(key);

    }

    /** Process the acknowledgement of an event's receipt by an entity */
    public void processAcknowledgement(long sequenceNumber, int entityId) throws
            ServiceException {
        Object key = new Long(sequenceNumber);

        SequenceDestinations seqDest = (SequenceDestinations) destinationTable
                                       .get(key);
        seqDest.removeDestination(entityId);

    }

    /** Retrieves a list of missed sequences between the specified range of
     sequence numbers. This method will return a null if there are indeed
     no missed sequences */
    public long[] retrieveMissedSequences(long startingAt, long endingAt,
                                          int templateId, int entityId) throws
            ServiceException {

        boolean done = false;
        long checkFor = startingAt;

        Vector _missing = new Vector();
        while (!done) {
            Object key = new Long(checkFor++);
            if (destinationTable.containsKey(key)) {
                SequenceDestinations _seqDest = (SequenceDestinations)
                                                destinationTable
                                                .get(key);
                if (_seqDest.containsDestination(entityId)) {
                    _missing.addElement(key);
                }
            }

            if (checkFor > endingAt) {
                done = true;
            }
        }

        int numMissing = _missing.size();
        if (numMissing == 0) {
            return null;
        }

        /** Proceed to copy the vector elements into a long array */
        long[] list = new long[numMissing];

        Enumeration e = _missing.elements();
        int tracker = 0;

        while (e.hasMoreElements()) {
            Long value = (Long) e.nextElement();
            list[tracker++] = value.longValue();
        }

        return list;
    }

    /** Advance the sync point associated with an entity */
    public void advanceSyncpoint(long syncpoint, int templateId, int entityId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        Object entityKey = new Integer(entityId);

        Hashtable entityList = (Hashtable) templatesAndSyncpoints.get(
                templateKey);
        if (entityList == null) {
            throw new ServiceException(moduleName + "The templateId=" +
                                       templateKey
                                       +
                    ", was not previously registered. Syncpoint"
                                       + " cannot be advanced for entity=" +
                                       entityId);
        }
        entityList.remove(entityKey);

        entityList.put(entityKey, new Long(syncpoint));

    }

    /** Get the syncpoint associated with an entity */
    public long getSyncpoint(int templateId, int entityId) throws
            ServiceException {

        Object templateKey = new Integer(templateId);
        Object entityKey = new Integer(entityId);

        Hashtable entityList = (Hashtable) templatesAndSyncpoints.get(
                templateKey);

        if (entityList == null) {
            throw new ServiceException(moduleName + "The templateId=" +
                                       templateKey
                                       +
                    ", was not previously registered. Syncpoint"
                                       + " cannot be retrieved for entity=" +
                                       entityId);
        }

        Long syncpoint = (Long) entityList.get(entityKey);

        if (syncpoint == null) {
            return 0;
        }

        return syncpoint.longValue();
    }

    /** Advance the catenation associated with an entity */
    public void advanceCatenation(int catenation, int templateId, int entityId) throws
            ServiceException {
        Object pubKey = templateId + ":" + entityId;
        Object catKey = new Integer(catenation);

        if (catenations.containsKey(pubKey)) {
            catenations.remove(catKey);
        }

        catenations.put(pubKey, catKey);
    }

    /** Get the catenation associated with an entity */
    public int getCatenation(int templateId, int entityId) throws
            ServiceException {
        Object pubKey = templateId + ":" + entityId;

        Integer catenation = (Integer) catenations.get(pubKey);

        if (catenation == null) {
            return 0;
        }

        return catenation.intValue();
    }

    /** Returns the sequence number last assigned by the reliable delivery
     service*/
    public long getSequenceNumberLastAssigned() throws ServiceException {
        return lastSequenceNumber;
    }

    /** Returns the sequence number last assigned by the reliable delivery
     service for a given templateId */
    public long getSequenceNumberLastAssigned(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        if (sequencesLastAssigned.containsKey(templateKey)) {
            return ((Long) sequencesLastAssigned.get(templateKey)).longValue();
        }
        return 0;
    }

    public long[] checkIfSequencesBelongToTemplate(int templateId,
            long[] sequences) throws ServiceException {

        return null;
    }

    public long[] retrieveSequences(long startingAt, long endingAt,
                                    int templateId, int maxOccur) throws
            ServiceException {
        return null;
    }

    public LinkedList retrieveSequencesAndTimestamps(int templateId) throws
            ServiceException {
        return null;
    }

    public LinkedList retrieveSequencesAndTimestamps(long startingTimestamp,
            long endingTimestamp, int templateId) throws ServiceException {
        return null;
    }

    public long retrieveTimestamp(int templateId, long sequenceNumber) throws
            ServiceException {
        return 0;
    }

    public LinkedList retrieveSequencesAndTimestampsStartingAt(
            long startingSequenceNumber, long endingTimestamp, int templateId) throws
            ServiceException {
        return null;
    }

    public LinkedList retrieveSequencesByTimestamps(long startingTimestamp,
            long endingTimestamp, int templateId) throws ServiceException {
        return null;
    }

    public long getTimestamp(long sequnceNumber, int templateId) throws
            ServiceException {
        return 0;
    }

    public LinkedList retrieveMinimumSequeceNumberAndTimestamp(int temlateId) throws
            ServiceException {
        return null;
    }

    public LinkedList retrieveMaximumSequeceNumberAndTimestamp(int temlateId) throws
            ServiceException {
        return null;
    }
}
